package com.niit.rdd

import org.apache.spark.{SparkConf, SparkContext}

object Spark_RDD_Action01 {

  def main(args: Array[String]): Unit = {
    //准备环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("ERROR")

    //行动算子
    /*
    reduce:聚集RDD中的所有元素，先聚合分内数据，再聚合分区间数据
     */
    val rdd1 =  sc.makeRDD( List(1,2,3,4),2 )//【1，2】 【3，4】
    val i:Int = rdd1.reduce(_+_)
    println(i)

    /*
    collect:在driver中，以数组的行为返回数据集的所有元素
     */
   val arrs1:Array[Int] = rdd1.collect();
   println(arrs1.mkString(","))

    /*
    count:获取RDD元素的个数
     */
    val co = rdd1.count()
    println(co)

    /*
    first:返回RDD中的第一个元素
     */
    val first =  rdd1.first();
    println(first)

    /*
    take: 返回一个由RDD的前n个元素组成的数组
     */
    val arrs2 =  rdd1.take(3)
    println(arrs2.mkString(","))

    /*
    takeOrdered ：返回该RDD排序后的前n个元素组成的数组
     */
    val rdd2 =  sc.makeRDD( List(2,1,4,3))
    val arrs3 =  rdd2.takeOrdered(2)
    println(arrs3.mkString(","))

    /*
    aggregate:分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的数据聚合
     */
    val res = rdd1.aggregate(10)(_+_,_+_)// 【1，2，10】13 【3，4，10】 17  +10
    println(res)

    val res2= rdd1.fold(10)(_+_)
    println(res2)

    /*
    countByKey:统计每种key出现的次数
     */
    val rdd3 = sc.makeRDD( List( ("a",1) ,("a",2),("b",3)  ) )

    val ckRdd =  rdd3.countByKey();
     println(ckRdd)

    /*
    save相关算子
      saveAsTextFile 任意类型
      saveAsObjectFile  任意类型
      saveAsSequenceFile数据格式必须是K-V类型
     */
    //rdd3.saveAsTextFile("output3")
    //rdd3.saveAsObjectFile("output4")//将数据转成对象字节码
    //要求数据格式必须是K-V类型
    rdd3.saveAsSequenceFile("output5")
    sc.stop()
  }

}
